package com.ddxz.flink.state

import org.apache.flink.core.fs.FileSystem
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

object ExampleCountWindowAvg extends App {
  val env = StreamExecutionEnvironment.getExecutionEnvironment
  env.fromCollection(List(
    (1L, 3L), (1L, 5L), (1L, 7L), (1L, 9L), (1L, 4L), (1L, 8L),
    (2L, 3L), (2L, 5L), (2L, 7L), (2L, 9L), (3L, 4L), (3L, 8L),
    (1L, 9L),(2L, 11L),(3L, 11L)
  )).keyBy(_._1).flatMap(new CollectStateAvg).writeAsCsv(".\\avg.csv", FileSystem.WriteMode.OVERWRITE)
  env.execute("avg")
}
